package com.example.openim.service.impl;

import com.example.openim.domain.OrgData;
import com.example.openim.domain.PersonData;
import com.example.openim.dto.PushRequestDto;
import com.example.openim.dto.RegisterRequestDto;
import com.example.openim.service.PushDataService;
import com.example.openim.utils.Md5Utils;
import com.example.openim.utils.PushCheckUtil;
import jakarta.servlet.http.HttpServletRequest;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.BulkOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.example.openim.constant.Constants.*;

/**
 * @description: 接收逻辑 - 组织 + 人员 均支持高并发唯一性校验
 * @author: xhq (final optimized)
 * @create: 2025-11-05
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class PushDataServiceImpl implements PushDataService {

    private final MongoTemplate mongoTemplate;
    private final RestTemplate restTemplate = new RestTemplate();

    // 注册接口并发控制
    private final Semaphore registerSemaphore = new Semaphore(20);

    // 第三方注册接口 URL
    @Value("${REGISTER.URL}")
    private String URL;

    @Autowired
    private Md5Utils md5Utils;

    @Autowired
    private PushCheckUtil pushCheckUtil;

    @Async("pushTaskExecutor")
    @Override
    public void processPushData(PushRequestDto dto, HttpServletRequest request) {
        try {
//            String data = request.getHeader("data");
//            String timestamp = request.getHeader("timestamp");
//            if (!pushCheckUtil.check(data,timestamp)) return;
            var root = dto.getRoot();
            if (root.getZzxx() != null && !root.getZzxx().isEmpty()) {
                log.info("开始处理组织信息");
                processOrgBatch(root.getZzxx());
            }
            if (root.getRyxx() != null && !root.getRyxx().isEmpty()) {
                log.info("开始处理人员信息");
                processPersonBatch(root.getRyxx());
            }
        } catch (Exception e) {
            log.error("推送处理失败", e);
        }
    }

    /**
     * 组织批量处理 - 支持更新（ZXZORG 为主键）
     */
    private void processOrgBatch(List<Map<String, Object>> list) {
        if (list.isEmpty()) return;

        BulkOperations bulkOps = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, OrgData.class);
        boolean hasUpsert = false;

        for (Map<String, Object> map : list) {
            String zxzorg = getStr(map, ZXZORG);
            if (zxzorg == null || zxzorg.isBlank()) continue;

            OrgData org = convertMapToOrg(map);
            Query query = new Query(Criteria.where(ZXZORG).is(zxzorg));
            Update update = buildOrgUpdate(org);
            bulkOps.upsert(query, update);
            hasUpsert = true;
        }

        if (hasUpsert) {
            try {
                bulkOps.execute();
                log.info("组织批量入库完成（支持更新），数量: {}", list.size());
            } catch (Exception e) {
                log.error("组织批量入库失败", e);
            }
        } else {
            log.info("组织批量入库跳过，无有效数据，原始数量: {}", list.size());
        }
    }

    /**
     * 人员数据批量处理 - 支持 ZSJHM 唯一性校验
     */
    private void processPersonBatch(List<Map<String, Object>> list) {
        if (list.isEmpty()) return;
        log.info("人员批量处理开始，请求参数：" + list);

        // 新增：收集所有 ZXZPER 用于查询现有人员数据
        Set<String> zxzperSet = list.stream()
                .map(m -> getStr(m, ZXZPER))
                .filter(Objects::nonNull)
                .collect(Collectors.toSet());

        // 新增：查询现有的人员数据
        Map<String, PersonData> existingPersonMap = new HashMap<>();
        if (!zxzperSet.isEmpty()) {
            Query query = new Query(Criteria.where(ZXZPER).in(zxzperSet));
            List<PersonData> existingPersons = mongoTemplate.find(query, PersonData.class, "person_data");
            existingPersonMap = existingPersons.stream()
                    .collect(Collectors.toMap(PersonData::getZxzper, Function.identity(), (oldVal, newVal) -> oldVal));
        }

        Set<String> batchPhones = list.stream()
                .map(m -> {
                    String phone = getStr(m, ZSJHM);
                    return phone == null ? null : phone.replace(" ", "");
                })
                .filter(phone -> phone != null && !phone.isBlank())
                .collect(Collectors.toSet());

        Map<String, String> dbPhoneToCodeMap = Collections.emptyMap();
        if (!batchPhones.isEmpty()) {
            Query query = new Query(Criteria.where("ZSJHM").in(batchPhones));
            List<PersonData> existingPersons = mongoTemplate.find(query, PersonData.class, "person_data");
            dbPhoneToCodeMap = existingPersons.stream()
                    .filter(p -> p.getZsjhm() != null && p.getZxzper() != null)
                    .collect(Collectors.toMap(
                            PersonData::getZsjhm,
                            PersonData::getZxzper,
                            (oldVal, newVal) -> oldVal
                    ));
        }

        BulkOperations bulkOps = mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED, "person_data");
        List<Runnable> registerTasks = new ArrayList<>();
        int skipped = 0;
        boolean hasUpsert = false;

        for (Map<String, Object> map : list) {
            String zxzper = getStr(map, ZXZPER);
            if (zxzper == null || zxzper.isBlank()) continue;

            PersonData person = convertMapToPerson(map);
            // 保存原始手机号用于注册
            String originalPhone = getStr(map, ZSJHM) == null ? null : getStr(map, ZSJHM).replace(" ", "");
            // 数据库存储带+86前缀的手机号
            String dbPhone = "+86 " + originalPhone;
            person.setZsjhm(dbPhone);

            if (originalPhone != null && !originalPhone.isBlank()) {
                String existingCode = dbPhoneToCodeMap.get(dbPhone);
                if (existingCode != null && !zxzper.equals(existingCode)) {
                    log.info("手机号冲突，跳过: {} (当前人员: {}, 已有人员: {})", dbPhone, zxzper, existingCode);
                    skipped++;
                    continue;
                }
            }

            // 新增：手机号校验和更新逻辑
            PersonData existingPerson = existingPersonMap.get(zxzper);
            if (existingPerson != null && existingPerson.getZsjhm() != null) {
                String dbMobileWithPrefix = existingPerson.getZsjhm(); // 数据库中的手机号(+86 xxx)
                // 去除 "+86 " 前缀进行比较
                if (dbMobileWithPrefix.startsWith("+86 ")) {
                    String dbMobile = dbMobileWithPrefix.substring(4); // 截取掉+86前缀

                    // 查找 credential 中 account 匹配 dbMobileWithPrefix 的记录
                    Query credentialQuery = new Query(Criteria.where("account").is(dbMobileWithPrefix));
                    Map<String, Object> credential = mongoTemplate.findOne(credentialQuery, Map.class, "credential");

                    if (credential != null && originalPhone != null) {
                        // 判断输入手机号是否与数据库中保存的手机号一致
                        if (!originalPhone.equals(dbMobile)) {
                            // 不一致则更新 credential 表中的 account 字段为 +86 + originalPhone
                            String newAccount = "+86 " + originalPhone;
                            Update update = new Update().set("account", newAccount);
                            Query updateQuery = new Query(Criteria.where("_id").is(credential.get("_id")));
                            mongoTemplate.updateFirst(updateQuery, update, "credential");
                            log.info("更新 credential 账户信息，人员编码: {}, 原手机号: {}, 新手机号: {}",
                                    zxzper, dbMobileWithPrefix, newAccount);
                        }
                    }
                }
            }
            // 新增：姓名更新逻辑
            String currentName = getStr(map, ZRYXM);  // 当前传入的姓名
            if (existingPerson != null) {
                String dbName = existingPerson.getZryxm();  // 数据库中的姓名

                if (currentName != null && !currentName.equals(dbName)) {
                    // 姓名发生变化，需要更新 user 表中的 nickname

                    // 先根据当前手机号查找 credential 获取 user_id
                    String currentPhoneWithPrefix = "+86 " + originalPhone;
                    Query credentialQuery = new Query(Criteria.where("account").is(currentPhoneWithPrefix));
                    Map<String, Object> credential = mongoTemplate.findOne(credentialQuery, Map.class, "credential");

                    if (credential != null && credential.get("user_id") != null) {
                        String userId = credential.get("user_id").toString();

                        // 更新 user 表中的 nickname 字段
                        Query userQuery = new Query(Criteria.where("user_id").is(userId));
                        Update userUpdate = new Update().set("nickname", currentName);
                        mongoTemplate.updateFirst(userQuery, userUpdate, "user");
                        log.info("更新 user 昵称信息，用户ID: {}, 人员编码: {}, 原姓名: {}, 新姓名: {}",
                                userId, zxzper, dbName, currentName);
                    }
                }
            }

            Query query = new Query(Criteria.where(ZXZPER).is(zxzper));
            Update update = buildPersonUpdate(person);
            bulkOps.upsert(query, update);
            hasUpsert = true;

            // 注册时使用原始手机号（不带+86）
            if (originalPhone != null && !originalPhone.isBlank()) {
                registerTasks.add(() -> registerSingleUser(person.getZryxm(), originalPhone));
            }
        }

        if (hasUpsert) {
            try {
                bulkOps.execute();
            } catch (Exception e) {
                log.error("人员批量入库失败", e);
            }
        } else {
            log.info("人员批量入库跳过，无有效数据（全部冲突或为空），原始数量: {}", list.size());
        }

        if (!registerTasks.isEmpty()) {
            executeRegisterTasks(registerTasks);
        }
    }


    /**
     * 异步执行注册任务（带信号量限流）
     */
    private void executeRegisterTasks(List<Runnable> tasks) {
        List<CompletableFuture<Void>> futures = tasks.stream()
                .map(task -> CompletableFuture.runAsync(() -> {
                    boolean acquired = false;
                    try {
                        acquired = registerSemaphore.tryAcquire(5, TimeUnit.SECONDS);
                        if (!acquired) {
                            log.warn("注册任务排队超时，等待执行");
                        }
                        task.run();
                    } catch (Exception e) {
                        log.debug("注册任务执行失败", e);
                    } finally {
                        if (acquired) {
                            registerSemaphore.release();
                        }
                    }
                }))
                .collect(Collectors.toList());

        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                .exceptionally(ex -> {
                    log.debug("注册任务批量异常: {}", ex.getMessage());
                    return null;
                });
    }

    /**
     * 单用户注册
     */
    private void registerSingleUser(String nickname, String phone) {
        try {
            RegisterRequestDto request = new RegisterRequestDto();
            RegisterRequestDto.User user = new RegisterRequestDto.User();
            user.setNickname(nickname);
            user.setPhoneNumber(phone);
            user.setPassword(md5Utils.lowMD5_32(PASSWORD));
            request.setUser(user);
            HttpHeaders headers = new HttpHeaders();
            headers.set("operationID", UUID.randomUUID().toString());
            headers.set("Content-Type", "application/json");
            HttpEntity<RegisterRequestDto> entity = new HttpEntity<>(request, headers);

            restTemplate.postForEntity(URL + REGISTER_URL, entity, String.class);
        } catch (Exception e) {
            log.warn("注册失败，手机号: {}", phone, e);
        }
    }

    // ====================== 工具方法 ======================

    private OrgData convertMapToOrg(Map<String, Object> m) {
        OrgData o = new OrgData();
        o.setZbmtx(getStr(m, ZBMTX));
        o.setZbmtxs(getStr(m, ZBMTXS));
        o.setZxzorg(getStr(m, ZXZORG));
        o.setZsjbmId(getStr(m, ZSJBMID));
        o.setZattr(getStr(m, ZATTR));
        o.setZbmpx(getStr(m, ZBMPX));
        o.setZbmde(getStr(m, ZBMDE));
        return o;
    }

    private PersonData convertMapToPerson(Map<String, Object> m) {
        PersonData p = new PersonData();
        p.setZryxm(getStr(m, ZRYXM));
        p.setZdname(getStr(m, ZDNAME));
        p.setZbmId(getStr(m, ZBMID));
        p.setZxzper(getStr(m, ZXZPER));
        p.setZrypx(getStr(m, ZRYPX));
        p.setZzgang(getStr(m, ZZGANG));
        p.setZxb(getStr(m, ZXB));
        p.setZsjhm(getStr(m, ZSJHM));
        p.setZdzyj(getStr(m, ZDZYJ));
        p.setZryde(getStr(m, ZRYDE));
        p.setZzwjb(getStr(m, ZZWJB));
        return p;
    }

    private String getStr(Map<String, Object> m, String k) {
        Object v = m.get(k);
        return v == null ? null : v.toString();
    }

    private Update buildOrgUpdate(OrgData o) {
        return new Update()
                .set(ZBMTX, o.getZbmtx())
                .set(ZBMTXS, o.getZbmtxs())
                .set(ZXZORG, o.getZxzorg())
                .set(ZSJBMID, o.getZsjbmId())
                .set(ZATTR, o.getZattr())
                .set(ZBMPX, o.getZbmpx())
                .set(ZBMDE, o.getZbmde())
                .set(BLOCK, o.getBlock());
    }

    private Update buildPersonUpdate(PersonData p) {
        return new Update()
                .set(ZRYXM, p.getZryxm())
                .set(ZDNAME, p.getZdname())
                .set(ZBMID, p.getZbmId())
                .set(ZXZPER, p.getZxzper())
                .set(ZRYPX, p.getZrypx())
                .set(ZZGANG, p.getZzgang())
                .set(ZXB, p.getZxb())
                .set(ZSJHM, p.getZsjhm())
                .set(ZDZYJ, p.getZdzyj())
                .set(ZRYDE, p.getZryde())
                .set(ZZWJB, p.getZzwjb())
                .set(BLOCK, p.getBlock());
    }

    /**
     * 使用集合差集方式处理兜底注册 - 更适合大数据量场景
     */
    public void fallbackRegisterUnregisteredUsers() {
        log.info("开始执行基于集合差集的兜底注册任务");

        try {
            // 分批处理，避免一次性加载大量数据到内存
            Query query = new Query();
            query.fields().include("Zsjhm", "Zryxm"); // 只查询需要的字段

            List<PersonData> allPersons = mongoTemplate.find(query, PersonData.class, "person_data");

            // 使用 groupingBy 而不是 toMap 避免丢失重复手机号的数据
            Map<String, List<PersonData>> phoneToPersonsMap = allPersons.stream()
                    .filter(person -> person.getZsjhm() != null && person.getZryxm() != null)
                    .collect(Collectors.groupingBy(PersonData::getZsjhm));

            log.info("总人员手机号数: {}, 去重后数量: {}", allPersons.size(), phoneToPersonsMap.size());

            // 获取已注册手机号集合
            List<Map> credentials = mongoTemplate.findAll(Map.class, "credential");
            Set<String> registeredPhones = credentials.stream()
                    .map(doc -> (String) doc.get("account"))
                    .filter(Objects::nonNull)
                    .collect(Collectors.toSet());

            log.info("已注册手机号数: {}", registeredPhones.size());

            // 计算差集：找出未注册的人员
            List<PersonData> unregisteredPersons = phoneToPersonsMap.entrySet().stream()
                    .filter(entry -> !registeredPhones.contains(entry.getKey()))
                    .flatMap(entry -> entry.getValue().stream())
                    .collect(Collectors.toList());

            log.info("待注册人员数: {}", unregisteredPersons.size());

            if (unregisteredPersons.isEmpty()) {
                log.info("没有需要注册的用户");
                return;
            }

            // 逐个注册未注册人员
            int successCount = 0;
            int failCount = 0;
            List<String> failedPhones = new ArrayList<>();

            for (PersonData person : unregisteredPersons) {
                String fullPhone = person.getZsjhm();
                if (fullPhone == null || !fullPhone.startsWith("+86 ")) {
                    log.warn("无效手机号格式跳过: {}", fullPhone);
                    continue;
                }

                String rawPhone = fullPhone.substring(4);

                try {
                    boolean success = fallbackRegisterSingleUser(person.getZryxm(), rawPhone);
                    if (success) {
                        successCount++;
                        log.debug("注册成功 - 昵称: {}, 手机号: {}", person.getZryxm(), rawPhone);
                    } else {
                        failCount++;
                        failedPhones.add(rawPhone);
                        log.warn("注册失败 - 昵称: {}, 手机号: {}", person.getZryxm(), rawPhone);
                    }
                } catch (Exception e) {
                    failCount++;
                    failedPhones.add(rawPhone);
                    log.warn("注册异常 - 昵称: {}, 手机号: {}", person.getZryxm(), rawPhone, e);
                }
            }

            // 输出最终统计结果
            log.info("兜底注册完成，成功: {} 条, 失败: {} 条", successCount, failCount);
            if (!failedPhones.isEmpty()) {
                //log.warn("失败手机号列表数量: {}", failedPhones.size());
                // 如果需要详细日志可以启用下面这行
                log.warn("失败手机号列表: {}", failedPhones);
            }

        } catch (Exception e) {
            log.error("兜底注册任务执行异常", e);
        }
    }


    /**
     * 单用户注册
     */
    private boolean fallbackRegisterSingleUser(String nickname, String phone) {
        try {
            RegisterRequestDto request = new RegisterRequestDto();
            RegisterRequestDto.User user = new RegisterRequestDto.User();
            user.setNickname(nickname);
            user.setPhoneNumber(phone);
            user.setPassword(md5Utils.lowMD5_32(PASSWORD));
            request.setUser(user);
            HttpHeaders headers = new HttpHeaders();
            headers.set("operationID", UUID.randomUUID().toString());
            headers.set("Content-Type", "application/json");
            HttpEntity<RegisterRequestDto> entity = new HttpEntity<>(request, headers);

            ResponseEntity<String> response = restTemplate.postForEntity(URL + REGISTER_URL, entity, String.class);

            // 可以添加对响应的检查，但保持原方法void返回类型
            if (response.getBody() != null) {
                log.info("注册接口返回参数：${}", response.getBody());
                if (!(response.getBody().contains("\"errCode\":0") || response.getBody().contains("\"errCode\": 0"))) {
                    return false;
                }
                return true;
            }
            return false;
        } catch (Exception e) {
            return false;
        }
    }

}
